package org.idea.eaglemq.broker;

import org.idea.eaglemq.broker.cache.CommonCache;
import org.idea.eaglemq.broker.config.ConsumeQueueOffsetLoader;
import org.idea.eaglemq.broker.config.EagleMqTopicLoader;
import org.idea.eaglemq.broker.config.GlobalPropertiesLoader;
import org.idea.eaglemq.broker.core.CommitLogAppendHandler;
import org.idea.eaglemq.broker.core.ConsumeQueueAppendHandler;
import org.idea.eaglemq.broker.core.ConsumeQueueConsumeHandler;
import org.idea.eaglemq.broker.model.EagleMqTopicModel;
import org.idea.eaglemq.broker.netty.broker.BrokerServer;
import org.idea.eaglemq.broker.rebalance.ConsumerInstancePool;

import java.io.IOException;


public class BrokerStartUp {

    private static GlobalPropertiesLoader globalPropertiesLoader;
    private static EagleMqTopicLoader eagleMqTopicLoader;
    private static CommitLogAppendHandler commitLogAppendHandler;
    private static ConsumeQueueOffsetLoader consumeQueueOffsetLoader;
    private static ConsumeQueueAppendHandler consumeQueueAppendHandler;
    private static ConsumeQueueConsumeHandler consumeQueueConsumeHandler;

    /**
     * 初始化配置逻辑
     */
    private static void initProperties() throws IOException {
        globalPropertiesLoader     = new GlobalPropertiesLoader();
        eagleMqTopicLoader         = new EagleMqTopicLoader();
        consumeQueueOffsetLoader   = new ConsumeQueueOffsetLoader();
        consumeQueueConsumeHandler = new ConsumeQueueConsumeHandler();
        commitLogAppendHandler     = new CommitLogAppendHandler();
        consumeQueueAppendHandler  = new ConsumeQueueAppendHandler();

        globalPropertiesLoader.loadProperties();
        eagleMqTopicLoader.loadProperties();
        eagleMqTopicLoader.startRefreshEagleMqTopicInfoTask();
        consumeQueueOffsetLoader.loadProperties();
        consumeQueueOffsetLoader.startRefreshConsumeQueueOffsetTask();
        for (EagleMqTopicModel eagleMqTopicModel : CommonCache.getEagleMqTopicModelMap().values()) {
            String topicName = eagleMqTopicModel.getTopic();
            commitLogAppendHandler.prepareMMapLoading(topicName);
            consumeQueueAppendHandler.prepareConsumeQueue(topicName);
        }
        CommonCache.setConsumeQueueConsumeHandler(consumeQueueConsumeHandler);
        CommonCache.setCommitLogAppendHandler(commitLogAppendHandler);
    }

    /**
     * 初始化和nameserver的长链接通道
     */
    private static void initNameServerChannel() {
        CommonCache.getNameServerClient().initConnection();
        CommonCache.getNameServerClient().sendRegistryMsg();
    }

    //开启重平衡任务
    private static void initReBalanceJob() {
        CommonCache.getConsumerInstancePool().startReBalanceJob();
    }

    private static void initBrokerServer() throws InterruptedException {
        BrokerServer brokerServer = new BrokerServer(CommonCache.getGlobalProperties().getBrokerPort());
        brokerServer.startServer();
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        //加载配置 ，缓存对象的生成
        initProperties();
        initNameServerChannel();
        initReBalanceJob();
        //这个函数是会阻塞的
        initBrokerServer();
    }
}

